package com.surfilter.massdata.spark.input;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.TreeMap;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.storage.StorageLevel;

import com.act.sparkanalyz.input.ISparkInput;
import com.act.sparkanalyz.service.impl.SparkService.ServiceContext;
import com.surfilter.massdata.spark.bean.DnsRegionHour;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;

public class HiveSparkInput implements ISparkInput {
    private static final Log log = LogFactory.getLog(HiveSparkInput.class);
    private static final long serialVersionUID = 1L;
    private String tableName;
	private String statisticalTime;
	private String sql;
	
    @Override
	public Map<String, DataFrame> read(ServiceContext serviceContext,
			Map<String, String> commandMap) throws Exception {
		log.info("============HiveSparkInput begin==============");
		Map<String, DataFrame> resultMap = new HashMap<String, DataFrame>();
		try{
			SQLContext sqlContext = serviceContext.getSqlCtx();
			HiveContext hiveCtx = (HiveContext) sqlContext;
			
			Broadcast<TreeMap<Long,Long>> ipBroadcast = serviceContext.getCtx().broadcast(CommonUtils.getIpExtensionMap());
			DataFrame dataFrame = null;
			if(StringUtils.equals(tableName, "d_dns_domain_hour")){
				dataFrame = hiveCtx.sql(getSourceSql(commandMap));
				JavaRDD<Row> sourceRDD = dataFrame.toJavaRDD();
				
				DataFrame cipDipFrame = getCipDipFilterDataFrame(sourceRDD,ipBroadcast,sqlContext);
				cipDipFrame.registerTempTable("cip_dip_filter_temp");
				cipDipFrame.persist(StorageLevel.MEMORY_AND_DISK_SER());
				
				DataFrame cipFrame = getCipFilterDataFrame(sourceRDD,ipBroadcast,sqlContext);
				cipFrame.registerTempTable("cip_filter_temp");
				cipFrame.persist(StorageLevel.MEMORY_AND_DISK_SER());
				
				DataFrame dipFrame = getDipFilterDataFrame(sourceRDD,ipBroadcast,sqlContext);
				dipFrame.registerTempTable("dip_filter_temp");
				dipFrame.show();
				dipFrame.persist(StorageLevel.MEMORY_AND_DISK_SER());
				
				resultMap.put("cip_dip_frame",cipDipFrame);
				resultMap.put("cip_frame",cipFrame);
				resultMap.put("dip_frame",dipFrame);
			}else{
				dataFrame = hiveCtx.sql(sql);
			}
			dataFrame.registerTempTable(tableName+"_temp");
			//持久化到内存中，若内存不足部分数据将会保存到磁盘中并序列化
			dataFrame.persist(StorageLevel.MEMORY_AND_DISK_SER());
			resultMap.put(tableName, dataFrame);
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		return resultMap;
	}

    @SuppressWarnings("serial")
	protected DataFrame getCipFilterDataFrame(JavaRDD<Row> sourceRDD,final Broadcast<TreeMap<Long,Long>> ipBroadcast,SQLContext sqlContext){
    	JavaRDD<DnsRegionHour> dnsRDD  = sourceRDD.filter(new Function<Row, Boolean>() {

			@Override
			public Boolean call(Row row){
				try{
					String cip = row.getAs("cip");
					if(CommonUtils.isExists(ipBroadcast.value(),cip)){
						return true;
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return false;
			}
		}).map(new Function<Row,DnsRegionHour>(){
			private static final long serialVersionUID = 1L;
			@Override
			public DnsRegionHour call(Row row) throws Exception {
				DnsRegionHour dns = new DnsRegionHour();
				String domain = row.getAs("domain");
				String cip = row.getAs("cip");
				String dip = row.getAs("dip");
				long dns_visit_count = row.getAs("dns_visit_count");
				
				dns.setDomain(domain);
				dns.setCip(cip);
				dns.setDip(dip);
				dns.setDns_visit_count(dns_visit_count);
				return dns;
			}});
    	
    	DataFrame df = sqlContext.createDataFrame(dnsRDD, DnsRegionHour.class);
    	return df;
    }
    
    @SuppressWarnings("serial")
    protected DataFrame getDipFilterDataFrame(JavaRDD<Row> sourceRDD,final Broadcast<TreeMap<Long,Long>> ipBroadcast,SQLContext sqlContext){
    	JavaRDD<DnsRegionHour> dnsRDD = sourceRDD.filter(new Function<Row, Boolean>() {

			@Override
			public Boolean call(Row row){
				try{
					String dip = row.getAs("dip");
					if(CommonUtils.isExists(ipBroadcast.value(),dip)){
						return true;
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return false;
			}
		}).map(new Function<Row,DnsRegionHour>(){
			private static final long serialVersionUID = 1L;
			@Override
			public DnsRegionHour call(Row row) throws Exception {
				DnsRegionHour dns = new DnsRegionHour();
				String domain = row.getAs("domain");
				String cip = row.getAs("cip");
				String dip = row.getAs("dip");
				long dns_visit_count = row.getAs("dns_visit_count");
				
				dns.setDomain(domain);
				dns.setCip(cip);
				dns.setDip(dip);
				dns.setDns_visit_count(dns_visit_count);
				return dns;
			}});
    	DataFrame df = sqlContext.createDataFrame(dnsRDD, DnsRegionHour.class);
    	return df;
    }
    
    @SuppressWarnings("serial")
    protected DataFrame getCipDipFilterDataFrame(JavaRDD<Row> sourceRDD,final Broadcast<TreeMap<Long,Long>> ipBroadcast,SQLContext sqlContext){
    	JavaRDD<DnsRegionHour> dnsRDD = sourceRDD.filter(new Function<Row, Boolean>() {

  			@Override
  			public Boolean call(Row row){
  				try{
  					TreeMap<Long,Long> ipExtensionMap = ipBroadcast.value();
  					String cip = row.getAs("cip");
  	  				String dip = row.getAs("dip");
  	  				if(CommonUtils.isExists(ipExtensionMap,cip) && CommonUtils.isExists(ipExtensionMap, dip)){
  	  					return true;
  	  				}
  				}catch(Exception e){
  					if(log.isErrorEnabled()){
  						log.error(e.getMessage(),e);
  					}
  				}
  				return false;
  			}
  		}).map(new Function<Row,DnsRegionHour>(){
			private static final long serialVersionUID = 1L;
			@Override
			public DnsRegionHour call(Row row) throws Exception {
				DnsRegionHour dns = new DnsRegionHour();
				String domain = row.getAs("domain");
				String cip = row.getAs("cip");
				String dip = row.getAs("dip");
				long dns_visit_count = row.getAs("dns_visit_count");
				
				dns.setDomain(domain);
				dns.setCip(cip);
				dns.setDip(dip);
				dns.setDns_visit_count(dns_visit_count);
				return dns;
			}});
      	DataFrame df = sqlContext.createDataFrame(dnsRDD, DnsRegionHour.class);
      	return df;
      }
    
	protected String getSourceSql(Map<String, String> commandMap){
		StringBuffer sqlBuffer = new StringBuffer();
		String dayStr = commandMap.get("-d");
		Date date = null;
		if(StringUtils.equals(statisticalTime, "hour")){
			date = DateUtil.getExecDate(DateUtil.getPreviousDayHour(new Date()),dayStr);
		}else{
			date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
		}
		String time = DateUtil.getCurrentymd(date);
		int hour = DateUtil.getCurrentHour(date);
		
		String hourStr = "";
		if(hour < 10){
			hourStr = "0" + hour;
		}else{
			hourStr = String.valueOf(hour);
		}
		sqlBuffer.append(sql);
		sqlBuffer.append(" where ds='" + time + "'");
		if(StringUtils.equals(statisticalTime, "hour")){
			sqlBuffer.append(" and hour='" + hourStr + "'");
		}
		return sqlBuffer.toString();
	}

	/*private DataFrame getDataFrame(JavaRDD<Row> sourceRDD,SQLContext sqlContext){
		JavaRDD<DnsRegionHour> rdd = sourceRDD.map(new Function<Row,DnsRegionHour>(){
			private static final long serialVersionUID = 1L;
			@Override
			public DnsRegionHour call(Row row) throws Exception {
				DnsRegionHour dns = new DnsRegionHour();
				String domain = row.getAs("domain");
				String cip = row.getAs("cip");
				long cip_value = DNSParser.ip2Long(cip);
				String dip = row.getAs("dip");
				long dip_value = DNSParser.ip2Long(dip);
				long dns_visit_count = row.getAs("dns_visit_count");
				
				dns.setDomain(domain);
				dns.setCip(cip);
				dns.setCip_value(cip_value);
				dns.setDip(dip);
				dns.setDip_value(dip_value);
				dns.setDns_visit_count(dns_visit_count);
				return dns;
			}});
		
		DataFrame df = sqlContext.createDataFrame(rdd, DnsRegionHour.class);
		return df;
	}*/
	
	/*//性能优于map操作，易出现OOM
	JavaRDD<DnsRegionHour> dnsRdd = dataFrame.toJavaRDD().mapPartitions(new FlatMapFunction<Iterator<Row>, DnsRegionHour>() {
		 
		@Override
		public Iterable<DnsRegionHour> call(Iterator<Row> it) throws Exception {
			List<DnsRegionHour> list = new ArrayList<DnsRegionHour>();
			while(it.hasNext()){
				Row row = it.next();
				DnsRegionHour dns = new DnsRegionHour();
				String domain = row.getAs("domain");
				String cip = row.getAs("cip");
				long cip_value = DNSParser.ip2Long(cip);
				String dip = row.getAs("dip");
				long dip_value = DNSParser.ip2Long(dip);
				long dns_visit_count = row.getAs("dns_visit_count");
				
				dns.setDomain(domain);
				dns.setCip(cip);
				dns.setCip_value(cip_value);
				dns.setDip(dip);
				dns.setDip_value(dip_value);
				dns.setDns_visit_count(dns_visit_count);
				list.add(dns);
			}
			return list;
		}
	});*/
}
